package service

import (
	"encoding/json"
	"github.com/mohae/deepcopy"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type ConvertTaskService struct {
	service.Service
}

func (e *ConvertTaskService) DeleteChannelTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.ChannelTask, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, liveService *liveService.LiveBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	//var m sync.Mutex
	liveBasicService.Orm = e.Orm
	//// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTask, 3)
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTask, wg *sync.WaitGroup) {
			//source := sourcesMap[v.SourceId]
			////删除自定义
			//if source.Kind == 2 && len(source.TaskId) > 0 {
			//	err := liveBasicService.DeleteCustomSource(source)
			//	if err != nil {
			//		e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			//删除协转任务流
			if len(v.TaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}
			if len(v.LiveId) > 0 {
				err := liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
				if err != nil {
					e.Log.Errorf("RemoveLive error:%s \r\n", err)
				}
			}
			err := e.Orm.Delete(&v).Error
			if err != nil {
				e.Log.Errorf("SaveWorkflow_Delete error:%s \r\n", err)
			}
			//m.Lock()
			//*partNum++
			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			//m.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

//CreateChannelTask 创建协转任务
//func (e *ConvertTaskService) CreateChannelTask(workTotalNum int, partNum *int, tasks []live_schedule.ChannelTask, upMap map[int]live_schedule.TrafficOn, offMap map[int]live_schedule.TrafficOff, basic live_schedule.ChannelBasic, mediaProcess live_schedule.MediaProcess, sourcesMap map[int]live_schedule.SourceBasic) error {
//
//	liveService := &liveService.LiveBasic{}
//	streamHubService := &streamHubService.StreamHubService{}
//	streamHubService.Log = e.Log
//	liveService.Log = e.Log
//	liveBasicService := SourceBasicService{}
//	liveBasicService.Log = e.Log
//	liveBasicService.Orm = e.Orm
//	staticsService := &ChannelStatisticsService{}
//	staticsService.Orm = e.Orm
//	staticsService.Log = e.Log
//	ids := make([]int, 0, 5)
//	for _, v := range tasks {
//		v := v
//		ids = append(ids, v.SourceId)
//	}
//	now := time.Now().Unix()
//	e.Orm.Model(&live_schedule.ChannelTask{})
//	// 创建任务通道
//	taskCh := make(chan live_schedule.ChannelTask, 3)
//	// 创建等待组
//	//var wg sync.WaitGroup
//	//wg.Add(len(tasks))
//	// 启动协程执行任务
//	go func() {
//		for i := 0; i < len(tasks); i++ {
//			taskCh <- tasks[i] // 将任务放入通道
//		}
//		close(taskCh) // 关闭通道，表示任务已经全部放入通道
//	}()
//	if len(taskCh) > 0 {
//		err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 4).Error
//		if err != nil {
//			return err
//		}
//	}
//
//	var m sync.Mutex
//	e.Log.Trace("信源信息为：", sourcesMap)
//	// 启动多个协程处理任务
//	for i := 0; i < 3; i++ {
//		go func() {
//			for v := range taskCh {
//				v := v
//				v1 := live_schedule.ChannelTask{}
//				//获取上车点
//				up := upMap[v.TrafficOnId]
//
//				//获取下车点配置
//				off := offMap[v.TrafficOffId]
//
//				//如果是srt拉流或者httpts走的还是内网
//				if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
//					off.Area = 2
//				}
//
//				//信源
//				source, ok := sourcesMap[v.SourceId]
//				if !ok {
//					//m.Unlock()
//					//*partNum++
//					//m.Unlock()
//					e.Log.Errorf("信源不存在")
//					continue
//				}
//				e.Log.Trace("当前使用信源：", source.SourceName)
//				//pullUrlPart := ""
//				pullDomain := ""
//				if (len(v.PushUrl) <= 0 && off.StreamType == 1) || (off.IsCustom == 1 && len(v.CustomOutUrl) <= 0) {
//					e.Log.Trace("不存在转推地址或者拉流地址")
//				} else {
//					req, autoReq, err := e.GenerateScheduleParam(up, off, mediaProcess, &v, source)
//					if err != nil {
//						e.Log.Errorf("请求参数失败")
//						continue
//					}
//					hub, err := streamHubService.CreateStreamHub(&req)
//					if err != nil {
//						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
//						//tx.Rollback()
//						//return err
//					}
//
//					if hub != nil {
//						if len(hub.Input.Sources) > 0 {
//							v.InUrl = hub.Input.Sources[0].UdpAddress
//							v.InNodeId = hub.Input.Sources[0].Id
//						}
//						//如果是外网拉流就不需要再次接收返回的输出url
//						if len(hub.Output.Targets) > 0 {
//							if !((off.Area == 1 || off.Area == 3) && off.StreamType == 2) {
//								v.OutUrl = hub.Output.Targets[0].Url
//							}
//							//如果下车点是httpts取AdditionalPublishUrl
//							if off.TrafficOffType == 3 {
//								v.OutUrl = hub.Output.Targets[0].AdditionalPublishUrl
//							}
//							v.OutNodeId = hub.Output.Targets[0].Id
//							if hub.Output.Targets[0].RedefinePlayUrlEnable == "1" {
//								path := hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
//								v.CustomOutUrl = path
//							}
//						}
//						v.StateUpdateTime = now
//						if basic.ChannelState == 2 {
//							v.TaskState = 3
//						}
//
//						v.TaskId = hub.TaskId
//						v.OutputStreamId = hub.OutputStreamName
//						v.UpdateTime = now
//					}
//
//					//自动调度内网情况
//					if off.Area == 3 && off.StreamType == 2 && (off.TrafficOffType == 1 || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
//						hub1, err := streamHubService.CreateStreamHub(&autoReq)
//						if err != nil {
//							e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
//							//tx.Rollback()
//							//return err
//						}
//						if hub1 != nil {
//							v1 = deepcopy.Copy(v).(live_schedule.ChannelTask)
//							if len(hub1.Input.Sources) > 0 {
//								v1.InUrl = hub1.Input.Sources[0].UdpAddress
//								v1.InNodeId = hub1.Input.Sources[0].Id
//							}
//
//							//如果是外网拉流就不需要再次接收返回的输出url
//							if len(hub1.Output.Targets) > 0 {
//								v1.OutUrl = hub1.Output.Targets[0].Url
//								pullDomain = v1.OutUrl[0:strings.LastIndex(v1.OutUrl, "/")]
//								v1.OutNodeId = hub1.Output.Targets[0].Id
//								if hub1.Output.Targets[0].RedefinePlayUrlEnable == "1" {
//									path := hub1.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub1.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
//									v1.CustomOutUrl = path
//								}
//							}
//							//如果下车点是httpts取AdditionalPublishUrl
//							if off.TrafficOffType == 3 {
//								v.OutUrl = hub1.Output.Targets[0].AdditionalPublishUrl
//							}
//							v.OutUrl = pullDomain + "/" + v.OutUrl[strings.LastIndex(v.OutUrl, "/")+1:]
//							v1.StateUpdateTime = now
//							if basic.ChannelState == 2 {
//								v1.TaskState = 3
//							}
//							v1.TaskId = hub1.TaskId
//							v1.Area = 2
//							v1.OutputStreamId = hub1.OutputStreamName
//							v1.UpdateTime = now
//						}
//					}
//
//				}
//				//如果通道状态是启用 就开启协助任务
//				e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
//				if basic.ChannelState == 1 {
//					if (source.Kind == 2 || source.Kind == 3) && len(source.TaskId) > 0 {
//						err := liveBasicService.StartSource(&dto.StartSourceReq{TenantId: basic.TenantId, Id: source.Id})
//						if err != nil {
//							e.Log.Errorf("StartStreamHub error:%s \r\n", err)
//							//tx.Rollback()
//							//return err
//						}
//					}
//					err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v.TaskId})
//					if err != nil {
//						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
//						//tx.Rollback()
//						//return err
//					}
//					//创建调度统计记录
//					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.TaskId, 1, now)
//					if err != nil {
//						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
//					}
//					if len(v1.TaskId) > 0 {
//						err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v1.TaskId})
//						if err != nil {
//							e.Log.Errorf("StartStreamHub error:%s \r\n", err)
//							//tx.Rollback()
//							//return err
//						}
//						//创建调度统计记录
//						err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v1.TaskId, 1, now)
//						if err != nil {
//							e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
//						}
//					}
//
//				}
//
//				//设置黑白名单
//				channelTaskService := ChannelTaskService{}
//				channelTaskService.Log = e.Log
//				taskIds := make([]string, 0, 3)
//				taskIds = append(taskIds, v.OutNodeId)
//				taskIds = append(taskIds, v1.OutNodeId)
//				channelTaskService.ProcessBlackWhite(taskIds, off)
//				m.Lock()
//				*partNum++
//				//channelTask := live_schedule.ChannelTask{}
//				//err := e.Orm.Model(&live_schedule.ChannelTask{}).Where("id=?", v.Id).Find(&channelTask).Error
//				//if err != nil {
//				//	e.Log.Errorf("get channel task error:%s \r\n", err)
//				//}
//				//
//				//if channelTask.TaskState != 3 {
//				//	v.TaskState = channelTask.TaskState
//				//	v1.TaskState = channelTask.TaskState
//				//}
//				err := e.Orm.Save(&v).Error
//				if err != nil {
//					e.Log.Errorf("Save channel task error:%s \r\n", err)
//				}
//				if len(v1.TaskId) > 0 {
//					err = e.Orm.Save(&v1).Error
//					if err != nil {
//						e.Log.Errorf("Save channel task error:%s \r\n", err)
//					}
//				}
//				//存储任务历史记录
//				if off.StreamType == 2 && len(v.TaskId) > 0 && len(v.OutUrl) > 0 {
//					history := live_schedule.ScheduleTaskHistory{}
//					history.TenantId = v.TenantId
//					history.ChannelId = v.ChannelId
//					history.PullUrl = v.OutUrl
//					history.OffId = v.TrafficOffId
//					history.CreateTime = now
//					history.UpdateTime = now
//					err = e.Orm.Save(&history).Error
//					if err != nil {
//						e.Log.Errorf("Save ScheduleTaskHistory error:%s \r\n", err)
//					}
//				}
//
//				//err = e.CalculateProgress(*partNum, workTotalNum, basic)
//				//if err != nil {
//				//	e.Log.Trace("CalculateProgress error：", err)
//				//}
//				e.Log.Trace("当前运行的任务个数：", *partNum, workTotalNum)
//				m.Unlock()
//
//			}
//		}()
//	}
//	return nil
//}
//

func (e *ConvertTaskService) CreateChannelTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.ChannelTask, upMap map[int]live_schedule.TrafficOn, offMap map[int]live_schedule.TrafficOff, basic live_schedule.ChannelBasic, mediaProcess live_schedule.MediaProcess, sourcesMap map[int]live_schedule.SourceBasic) error {

	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	liveBasicService := SourceBasicService{}
	liveBasicService.Log = e.Log
	liveBasicService.Orm = e.Orm
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	ids := make([]int, 0, 5)
	for _, v := range tasks {
		v := v
		ids = append(ids, v.SourceId)
	}
	now := time.Now().Unix()
	e.Orm.Model(&live_schedule.ChannelTask{})
	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTask, 3)
	// 创建等待组

	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//if len(taskCh) > 0 {
	//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 4).Error
	//	if err != nil {
	//		return err
	//	}
	//}

	var m sync.Mutex
	e.Log.Trace("信源信息为：", sourcesMap)
	// 启动多个协程处理任务
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTask, wg *sync.WaitGroup) {
			defer wg.Done()
			v1 := live_schedule.ChannelTask{}
			//获取上车点
			up := upMap[v.TrafficOnId]

			//获取下车点配置
			off := offMap[v.TrafficOffId]

			//如果是srt拉流或者httpts走的还是内网
			if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
				off.Area = 2
			}

			//信源
			source, ok := sourcesMap[v.SourceId]
			if !ok {
				//m.Unlock()
				//*partNum++
				//m.Unlock()
				e.Log.Errorf("信源不存在")
				return
			}
			e.Log.Trace("当前使用信源：", source.SourceName)
			//pullUrlPart := ""
			pullDomain := ""
			if (len(v.PushUrl) <= 0 && off.StreamType == 1) || (off.IsCustom == 1 && len(v.CustomOutUrl) <= 0) {
				e.Log.Trace("不存在转推地址或者拉流地址")
			} else {
				req, autoReq, err := e.GenerateScheduleParam(up, off, mediaProcess, &v, source)
				if err != nil {
					e.Log.Errorf("请求参数失败")
					return
				}
				hub, err := streamHubService.CreateStreamHub(&req)
				if err != nil {
					e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}

				if hub != nil {
					if len(hub.Input.Sources) > 0 {
						v.InUrl = hub.Input.Sources[0].UdpAddress
						v.InNodeId = hub.Input.Sources[0].Id
					}
					//如果是外网拉流就不需要再次接收返回的输出url
					if len(hub.Output.Targets) > 0 {
						if !((off.Area == 1 || off.Area == 3) && off.StreamType == 2) {
							v.OutUrl = hub.Output.Targets[0].Url
						}
						//如果下车点是httpts取AdditionalPublishUrl
						if off.TrafficOffType == 3 {
							v.OutUrl = hub.Output.Targets[0].AdditionalPublishUrl
						}
						v.OutNodeId = hub.Output.Targets[0].Id
						if hub.Output.Targets[0].RedefinePlayUrlEnable == "1" {
							path := hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
							v.CustomOutUrl = path
						}
					}
					v.StateUpdateTime = now
					if basic.ChannelState == 2 {
						v.TaskState = 3
					}

					v.TaskId = hub.TaskId
					v.OutputStreamId = hub.OutputStreamName
					v.UpdateTime = now
				}

				//自动调度内网情况
				if off.Area == 3 && off.StreamType == 2 && (off.TrafficOffType == 1 || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
					hub1, err := streamHubService.CreateStreamHub(&autoReq)
					if err != nil {
						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					if hub1 != nil {
						v1 = deepcopy.Copy(v).(live_schedule.ChannelTask)
						if len(hub1.Input.Sources) > 0 {
							v1.InUrl = hub1.Input.Sources[0].UdpAddress
							v1.InNodeId = hub1.Input.Sources[0].Id
						}

						//如果是外网拉流就不需要再次接收返回的输出url
						if len(hub1.Output.Targets) > 0 {
							v1.OutUrl = hub1.Output.Targets[0].Url
							pullDomain = v1.OutUrl[0:strings.LastIndex(v1.OutUrl, "/")]
							v1.OutNodeId = hub1.Output.Targets[0].Id
							if hub1.Output.Targets[0].RedefinePlayUrlEnable == "1" {
								path := hub1.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub1.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
								v1.CustomOutUrl = path
							}
						}
						//如果下车点是httpts取AdditionalPublishUrl
						if off.TrafficOffType == 3 {
							v.OutUrl = hub1.Output.Targets[0].AdditionalPublishUrl
						}
						v.OutUrl = pullDomain + "/" + v.OutUrl[strings.LastIndex(v.OutUrl, "/")+1:]
						v1.StateUpdateTime = now
						if basic.ChannelState == 2 {
							v1.TaskState = 3
						}
						v1.TaskId = hub1.TaskId
						v1.Area = 2
						v1.OutputStreamId = hub1.OutputStreamName
						v1.UpdateTime = now
					}
				}

			}
			//如果通道状态是启用 就开启协助任务
			e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
			if basic.ChannelState == 1 {
				if (source.Kind == 2 || source.Kind == 3) && len(source.TaskId) > 0 {
					err := liveBasicService.StartSource(&dto.StartSourceReq{TenantId: basic.TenantId, Id: source.Id})
					if err != nil {
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
				}
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
				if len(v1.TaskId) > 0 {
					err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v1.TaskId})
					if err != nil {
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					//创建调度统计记录
					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v1.TaskId, 1, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
				}

			}

			//设置黑白名单
			channelTaskService := ChannelTaskService{}
			channelTaskService.Log = e.Log
			taskIds := make([]string, 0, 3)
			taskIds = append(taskIds, v.OutNodeId)
			taskIds = append(taskIds, v1.OutNodeId)
			channelTaskService.ProcessBlackWhite(taskIds, off)
			m.Lock()
			*partNum++
			//channelTask := live_schedule.ChannelTask{}
			//err := e.Orm.Model(&live_schedule.ChannelTask{}).Where("id=?", v.Id).Find(&channelTask).Error
			//if err != nil {
			//	e.Log.Errorf("get channel task error:%s \r\n", err)
			//}
			//
			//if channelTask.TaskState != 3 {
			//	v.TaskState = channelTask.TaskState
			//	v1.TaskState = channelTask.TaskState
			//}
			err := e.Orm.Save(&v).Error
			if err != nil {
				e.Log.Errorf("Save channel task error:%s \r\n", err)
			}
			if len(v1.TaskId) > 0 {
				err = e.Orm.Save(&v1).Error
				if err != nil {
					e.Log.Errorf("Save channel task error:%s \r\n", err)
				}
			}
			//存储任务历史记录
			if off.StreamType == 2 && len(v.TaskId) > 0 && len(v.OutUrl) > 0 {
				history := live_schedule.ScheduleTaskHistory{}
				history.TenantId = v.TenantId
				history.ChannelId = v.ChannelId
				history.PullUrl = v.OutUrl
				history.OffId = v.TrafficOffId
				history.CreateTime = now
				history.UpdateTime = now
				err = e.Orm.Save(&history).Error
				if err != nil {
					e.Log.Errorf("Save ScheduleTaskHistory error:%s \r\n", err)
				}
			}

			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			e.Log.Trace("当前运行的任务个数：", *partNum, workTotalNum)
			m.Unlock()

		}(tasks[i], wg)
	}
	return nil
}

func (e *ConvertTaskService) GenerateScheduleParam(up live_schedule.TrafficOn, off live_schedule.TrafficOff, mediaProcess live_schedule.MediaProcess, task *live_schedule.ChannelTask, source live_schedule.SourceBasic) (StreamHubDto.StreamHubReq, StreamHubDto.StreamHubReq, error) {

	pullUrlPart := ""
	liveService := &liveService.LiveBasic{}
	liveService.Log = e.Log
	target := StreamHubDto.Target{}
	target.Format = protocols[off.TrafficOffType]
	target.Id = task.OutNodeId
	//输入参数
	sourceParam := StreamHubDto.Source{}

	//使用转码源还是原始源
	if up.StreamType == 1 {
		e.Log.Trace("当前使用原始源：", up.SourceId)
		sourceParam.Id = task.InNodeId
		sourceParam.Url = source.OutUdpUrl
	} else {
		sourceParam.Id = task.InNodeId
		sourceParam.Url = up.TransUdpUrl
	}

	req := StreamHubDto.StreamHubReq{}

	if len(task.TaskId) > 0 {
		req.Id = task.TaskId
	}

	req.Name = ""
	req.Type = "live"
	req.TenantId = task.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[5],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    "push",
		Sources: []StreamHubDto.Source{sourceParam},
	}

	//第三方拉流+内网 输出类型为pull 其他情况为push
	if off.StreamType == 2 && off.Area == 2 {
		target.Type = "pull"
	} else {
		target.Type = "push"
	}
	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	if mediaProcess.PlayType == 2 {
		delay.Enable = "yes"
		delay.Duration = mediaProcess.Delay * 1000
	}
	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/convert-task-callback"

	target1 := deepcopy.Copy(target).(StreamHubDto.Target)
	target1.Type = "pull"
	//外网+第三方拉流需要先调用全球直播
	if (off.Area == 3 || off.Area == 1) && off.StreamType == 2 {
		target.Format = "rtmp"
		streamReq := e.GenerateInsertLiveParam(task.TenantId)
		res, err := liveService.CreateLiveStream(streamReq)
		if err != nil {
			e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
			return req, req, err
		}

		task.LiveId = res.LiveId
		//记录额外信息
		extraMap := make(map[string]string)
		extraMap["rtmp"] = res.RtmpUrl
		extraMap["hls"] = res.HlsUrl
		extraMap["flv"] = res.FlvUrl

		data, err := json.Marshal(extraMap)
		if err != nil {
			e.Log.Errorf("Marshal error:%s \r\n", err)
			return req, req, err
		}
		task.ExtraInfo = string(data)
		pushes := strings.Split(res.PushUrl, models.Delimiter)
		if len(pushes) > 0 {
			target.Url = pushes[0]
			task.PushUrl = pushes[0]
		}

		if off.TrafficOffType == 9 {
			off.TrafficOffType = 1
		}
		task.OutUrl = extraMap[protocols[off.TrafficOffType]]

		u, err := url.Parse(task.OutUrl)
		if err != nil {
			return req, req, err
		}
		path := strings.TrimPrefix(u.Path, "/")
		streamName := ""
		if len(path) > 1 {
			streamName = strings.Split(path, "/")[1]
		}

		queryString := strings.Split(streamName, ".")[0]
		pullUrlPart = queryString
		//如果是外网并且拉流地址
		if off.Area == 3 && off.StreamType == 2 {
			task.Area = 1
		}

	} else if off.StreamType == 1 {
		target.Url = task.PushUrl
	}
	target.RedefinePlayUrlEnable = "2"
	params := StreamHubDto.RedefinePlayUrlParams{}
	target.RedefinePlayUrlParams = params

	//自动调度请求参数
	autoReq := deepcopy.Copy(req).(StreamHubDto.StreamHubReq)
	//如果是自定义拉流地址
	if off.StreamType == 2 && off.TrafficOffType == 2 && (off.Area == 2 || off.Area == 3) && off.IsCustom == 1 {
		arr := strings.Split(task.CustomOutUrl, "/")
		if off.Area == 3 {
			if len(arr) > 1 {
				target1.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target1.RedefinePlayUrlParams = params
			}
		} else {
			if len(arr) > 1 {
				target.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target.RedefinePlayUrlParams = params
			}
		}

	}
	//创建协助任务
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target1},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.OutputStreamName = pullUrlPart
	return req, autoReq, nil
}

func (e *ConvertTaskService) StartChannelTasks(wg *sync.WaitGroup, tasks []live_schedule.ChannelTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//var m sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTask, wg *sync.WaitGroup) {
			defer wg.Done()
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, TaskId: up.TaskId, Id: v.SourceId})
			//	if err != nil {
			//		e.Log.Errorf("StartSource error:%s \r\n", err)
			//	}
			//}
			if len(v.TaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = channelStatisticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
				}
			}
			//m.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//param["progress"] = formattedResult
			//if workNum <= *partNum {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//} else {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//m.Unlock()
		}(tasks[i], wg)
	}
	return nil

}

//func (e *ConvertTaskService) StartChannelTasks(tasks []live_schedule.ChannelTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
//	// 创建任务通道
//	taskCh := make(chan live_schedule.ChannelTask, 3)
//	streamHubService := &streamHubService.StreamHubService{}
//	streamHubService.Log = e.Log
//	sourceBasicService := &SourceBasicService{}
//	sourceBasicService.Log = e.Log
//	sourceBasicService.Orm = e.Orm
//	channelStatisticsService := &ChannelStatisticsService{}
//	channelStatisticsService.Log = e.Log
//	channelStatisticsService.Orm = e.Orm
//	// 启动协程执行任务
//	go func() {
//		for i := 0; i < len(tasks); i++ {
//			taskCh <- tasks[i] // 将任务放入通道
//		}
//		close(taskCh) // 关闭通道，表示任务已经全部放入通道
//	}()
//	//var m sync.Mutex
//	now := time.Now().Unix()
//	for i := 0; i < 3; i++ {
//		go func() {
//			for v := range taskCh {
//				//up := sourceMap[v.SourceId]
//				//if up.Kind == 2 {
//				//	err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, TaskId: up.TaskId, Id: v.SourceId})
//				//	if err != nil {
//				//		e.Log.Errorf("StartSource error:%s \r\n", err)
//				//	}
//				//}
//				if len(v.TaskId) > 0 {
//					//开启协助任务
//					err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
//					if err != nil {
//						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
//						//tx.Rollback()
//						//return err
//					}
//					//创建调度统计记录
//					err = channelStatisticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TaskId, 1, now)
//					if err != nil {
//						e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
//					}
//				}
//				//m.Lock()
//				//*partNum++
//				//result := float64(*partNum) / float64(workNum)
//				//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
//				//param["progress"] = formattedResult
//				//if workNum <= *partNum {
//				//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
//				//	if err != nil {
//				//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
//				//	}
//				//} else {
//				//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
//				//	if err != nil {
//				//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
//				//	}
//				//}
//				//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
//				//m.Unlock()
//			}
//		}()
//	}
//	return nil
//
//}
//

func (e *ConvertTaskService) StopChannelTasks(wg *sync.WaitGroup, tasks []live_schedule.ChannelTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	//var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTask, wg *sync.WaitGroup) {
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
			//	if err != nil {
			//		e.Log.Errorf("StopSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			//停止协助任务
			if len(v.TaskId) > 0 {
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)
				}
			}
			//mu.Lock()
			//更新通道任务状态为离线
			param1 := make(map[string]interface{})
			param1["update_time"] = now
			param1["task_state"] = 3
			err := e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=? and task_id=?", v.TenantId, v.ChannelId, v.TaskId).Updates(&param1).Error
			if err != nil {
				e.Log.Errorf("ProcessWorkflow Updates ChannelTask error:%s \r\n", err)
			}
			//更新任务记录结束时间
			param2 := make(map[string]interface{})
			param2["end_time"] = now
			err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", v.TaskId).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			}
			//
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//param["progress"] = formattedResult
			//if *partNum == workNum {
			//	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
			//	}
			//} else {
			//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//mu.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

//func (e *ConvertTaskService) StopChannelTasks(tasks []live_schedule.ChannelTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
//	// 创建任务通道
//	taskCh := make(chan live_schedule.ChannelTask, 3)
//	streamHubService := &streamHubService.StreamHubService{}
//	streamHubService.Log = e.Log
//	sourceBasicService := &SourceBasicService{}
//	sourceBasicService.Log = e.Log
//	sourceBasicService.Orm = e.Orm
//	// 启动协程执行任务
//	go func() {
//		for i := 0; i < len(tasks); i++ {
//			taskCh <- tasks[i] // 将任务放入通道
//		}
//		close(taskCh) // 关闭通道，表示任务已经全部放入通道
//	}()
//
//	//var mu sync.Mutex
//	now := time.Now().Unix()
//	for i := 0; i < 3; i++ {
//		go func() {
//			for v := range taskCh {
//				//up := sourceMap[v.SourceId]
//				//if up.Kind == 2 {
//				//	err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
//				//	if err != nil {
//				//		e.Log.Errorf("StopSource error:%s \r\n", err)
//				//	}
//				//}
//				//停止协助任务
//				if len(v.TaskId) > 0 {
//					err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
//					if err != nil {
//						e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)
//					}
//				}
//				//mu.Lock()
//				//更新通道任务状态为离线
//				param1 := make(map[string]interface{})
//				param1["update_time"] = now
//				param1["task_state"] = 3
//				err := e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=? and task_id=?", v.TenantId, v.ChannelId, v.TaskId).Updates(&param1).Error
//				if err != nil {
//					e.Log.Errorf("ProcessWorkflow Updates ChannelTask error:%s \r\n", err)
//				}
//				//更新任务记录结束时间
//				param2 := make(map[string]interface{})
//				param2["end_time"] = now
//				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", v.TaskId).Updates(&param2).Error
//				if err != nil {
//					e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
//				}
//				//
//				//*partNum++
//				//result := float64(*partNum) / float64(workNum)
//				//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
//				//param["progress"] = formattedResult
//				//if *partNum == workNum {
//				//	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
//				//	if err != nil {
//				//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
//				//	}
//				//} else {
//				//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
//				//	if err != nil {
//				//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
//				//	}
//				//}
//				//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
//				//mu.Unlock()
//			}
//		}()
//	}
//
//	return nil
//
//}

func (e *ConvertTaskService) GenerateInsertLiveParam(tenantId string) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = ""
	streamReq.TenantId = tenantId
	streamReq.ExpireTime = 10 * 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

func (e *ConvertTaskService) ConvertTaskCallBack(c *dto.TargetCallBackReq) error {

	//更新通道状态
	param := make(map[string]interface{})
	if c.TransferState == "Capturing" {
		param["task_state"] = 1
	} else {
		param["task_state"] = 2
	}
	param["state_update_time"] = c.TransferStateLastUpdateTime
	e.Log.Trace("阿斯顿好：", param)
	err := e.Orm.Model(live_schedule.CovertTask{}).Where("task_id=? and out_node_id=?", c.Id, c.TaskId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("ConvertTaskCallBack error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ConvertTaskService) CalculateProgress(partNum int, workTotalNum int, basic live_schedule.ChannelBasic) error {
	result := float64(partNum) / float64(workTotalNum)
	formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
	e.Log.Trace("当前进度为", formattedResult)
	param := make(map[string]interface{})
	param["progress"] = formattedResult
	if workTotalNum == partNum {
		param["channel_state"] = basic.ChannelState
	}
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	if err != nil {
		e.Log.Trace("update ChannelBasic error：", err)
	}
	return err
}

func (e *ConvertTaskService) GenerateScheduleParam1(node live_schedule.Node, task *live_schedule.CovertTask, inputModels []live_schedule.InputModel) (StreamHubDto.StreamHubReq, error) {

	//1rtmp 2flv 3hls 4https 5 srt 6udp 7flv 8sdi 9外场设备
	pullUrlPart := ""
	liveService := &liveService.LiveBasic{}
	liveService.Log = e.Log
	target := StreamHubDto.Target{}
	target.Format = node.Data.Protocol
	target.Id = task.OutNodeId
	//输入参数
	sourceParam := StreamHubDto.Source{}

	sourceParam.Id = task.InNodeId
	sourceParam.Url = inputModels[0].InputUrl

	req := StreamHubDto.StreamHubReq{}

	if len(task.TaskId) > 0 {
		req.Id = task.TaskId
		target.Url = task.PushUrl
	}

	req.Name = ""
	req.Type = "live"
	req.TenantId = task.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[5],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    "push",
		Sources: []StreamHubDto.Source{sourceParam},
	}
	//如果是httpts的话走内网
	if target.Format == "httpts" {
		node.Data.Area = 2
	}

	//第三方拉流+内网 输出类型为pull 其他情况为push
	if node.Data.StreamType == 2 && node.Data.Area == 2 {
		target.Type = "pull"
	} else {
		target.Type = "push"
	}
	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}

	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/convert-task-callback"

	target1 := deepcopy.Copy(target).(StreamHubDto.Target)
	target1.Type = "pull"
	//外网+第三方拉流需要先调用全球直播
	if (node.Data.Area == 3 || node.Data.Area == 1) && node.Data.StreamType == 2 && len(task.TaskId) <= 0 {
		target.Format = "rtmp"
		streamReq := e.GenerateInsertLiveParam(task.TenantId)
		res, err := liveService.CreateLiveStream(streamReq)
		if err != nil {
			e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
			return req, err
		}

		task.LiveId = res.LiveId
		//记录额外信息
		extraMap := make(map[string]string)
		extraMap["rtmp"] = res.RtmpUrl
		extraMap["hls"] = res.HlsUrl
		extraMap["flv"] = res.FlvUrl

		data, err := json.Marshal(extraMap)
		if err != nil {
			e.Log.Errorf("Marshal error:%s \r\n", err)
			return req, err
		}
		task.ExtraInfo = string(data)
		pushes := strings.Split(res.PushUrl, models.Delimiter)
		if len(pushes) > 0 {
			target.Url = pushes[0]
			task.PushUrl = pushes[0]
		}

		//if node.Protocol == 9 {
		//	off.TrafficOffType = 1
		//}
		task.OutUrl = extraMap[node.Data.Protocol]

		u, err := url.Parse(task.OutUrl)
		if err != nil {
			return req, err
		}
		path := strings.TrimPrefix(u.Path, "/")
		streamName := ""
		if len(path) > 1 {
			streamName = strings.Split(path, "/")[1]
		}

		queryString := strings.Split(streamName, ".")[0]
		pullUrlPart = queryString
		//如果是外网并且拉流地址
		if node.Data.Area == 3 && node.Data.StreamType == 2 {
			task.Area = 1
		}

	} else if node.Data.StreamType == 1 {
		target.Url = node.Data.SourceUrl
	}

	target.RedefinePlayUrlEnable = "2"
	params := StreamHubDto.RedefinePlayUrlParams{}
	target.RedefinePlayUrlParams = params

	//自动调度请求参数
	autoReq := deepcopy.Copy(req).(StreamHubDto.StreamHubReq)
	//如果是自定义拉流地址
	if node.Data.StreamType == 2 && node.Data.Protocol == "hls" && (node.Data.Area == 2 || node.Data.Area == 3) && node.Data.IsCustom == 1 {
		task.CustomOutUrl = node.Data.CustomUrl
		arr := strings.Split(task.CustomOutUrl, "/")
		if node.Data.Area == 3 {
			if len(arr) > 1 {
				target1.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target1.RedefinePlayUrlParams = params
			}
		} else {
			if len(arr) > 1 {
				target.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target.RedefinePlayUrlParams = params
			}
		}

	}
	//创建协助任务
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target1},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.OutputStreamName = pullUrlPart
	return req, nil
}

func (e *ConvertTaskService) CreateChannelTask1(node live_schedule.Node, task *live_schedule.CovertTask, inputModels []live_schedule.InputModel, basic live_schedule.ChannelBasic) (live_schedule.CovertTask, error) {

	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	liveBasicService := SourceBasicService{}
	liveBasicService.Log = e.Log
	liveBasicService.Orm = e.Orm
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log

	now := time.Now().Unix()
	req, err := e.GenerateScheduleParam1(node, task, inputModels)
	if err != nil {
		e.Log.Errorf("请求参数失败")
		return *task, err
	}
	if len(task.TaskId) <= 0 {

		hub, err := streamHubService.CreateStreamHub(&req)
		if err != nil {
			e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
			//tx.Rollback()
			return *task, err
		}
		task.MediaId = node.Id
		if hub != nil {
			task.TaskId = hub.TaskId
			if len(hub.Input.Sources) > 0 {
				task.InUrl = hub.Input.Sources[0].UdpAddress
				task.InNodeId = hub.Input.Sources[0].Id
			}
			//如果是外网拉流就不需要再次接收返回的输出url
			if len(hub.Output.Targets) > 0 {
				if !((node.Data.Area == 1 || node.Data.Area == 3) && node.Data.StreamType == 2) {
					task.OutUrl = hub.Output.Targets[0].Url
				}
				//如果下车点是httpts取AdditionalPublishUrl
				if node.Data.Protocol == "httpts" {
					task.OutUrl = hub.Output.Targets[0].AdditionalPublishUrl
				}
				task.OutNodeId = hub.Output.Targets[0].Id
				if hub.Output.Targets[0].RedefinePlayUrlEnable == "1" {
					path := hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
					task.CustomOutUrl = path
				}
			}
			task.StateUpdateTime = now
			//if basic.ChannelState == 2 {
			//	task.TaskState = 3
			//}

		}
	} else {
		_, err = streamHubService.UpdateStreamHub(&req)
		if err != nil {
			e.Log.Errorf("UpdateStreamHub error:%s \r\n", err)
			//tx.Rollback()
			return *task, err
		}
	}

	err = e.Orm.Save(&task).Error
	if err != nil {
		e.Log.Errorf("Save ChannelTask error:", err)
		return *task, err
	}

	if basic.ChannelState != 2 {
		err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: task.TaskId})
		if err != nil {
			e.Log.Errorf("StartStreamHub error:%s \r\n", err)
			return *task, err
		}

		//创建调度统计记录
		err = channelStatisticsService.CreateScheduleRecord(task.TenantId, task.ChannelId, task.TaskId, 1, now)
		if err != nil {
			e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
		}
	}

	if node.Data.EnableBlackWhite == 1 {
		channelTaskService := ChannelTaskService{}
		channelTaskService.Log = e.Log
		taskIds := make([]string, 0, 3)
		taskIds = append(taskIds, task.OutNodeId)
		err = channelTaskService.ProcessBlackWhite1(taskIds, task.TenantId, node)
		if err != nil {
			e.Log.Errorf("ProcessBlackWhite1 error:%s \r\n", err)
			return *task, err
		}
		hlsBlackWhites := make([]live_schedule.HlsBlackWhite, 0, 5)
		for _, address := range node.Data.IPAddresses {
			address := address
			hlsBlackWhite := live_schedule.HlsBlackWhite{}
			hlsBlackWhite.TenantId = task.TenantId
			hlsBlackWhite.IpAddress = address
			hlsBlackWhite.ChannelId = task.ChannelId
			hlsBlackWhite.NodeId = node.Id
			hlsBlackWhites = append(hlsBlackWhites, hlsBlackWhite)
		}
		err = e.Orm.Save(&hlsBlackWhites).Error
		if err != nil {
			e.Log.Errorf("save hlsBlackWhites error:%s \r\n", err)
			return *task, err
		}
	}
	return *task, nil
}
